package com.gxy.learn.thread.simple;

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 多线程 队列等待
 * author: Gao xueyong
 * Create at: 2022/3/19 下午6:23
 */
@Slf4j
@Component
public class WaitBlockedPool {

    private int count = 0;
    ReentrantLock lock = new ReentrantLock();

    /**
     * 根据自定义阻塞策略来达到当队列满时进行阻塞进而保持任务持续运行的目的
     */
    @Scheduled(cron = "0 0/1 * * * ?")
    public void blockingQueueDefaultTask() {
        ExecutorService threadPool = new ThreadPoolExecutor(5, 10,
                0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(10), new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                try {
                    log.info("队列阻塞中...");
                    executor.getQueue().put(r);
                } catch (InterruptedException e) {
                    log.error("err ", e);
                }
            }
        });

        int i = 0;
        while (i < 100) {
            i++;
            int finalI = i;
            threadPool.submit(new Runnable() {
                @Override
                public void run() {
                    log.info("当前任务 {}", finalI);
                    increment();
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        log.error("err ", e);
                    }
                    if (finalI == 99) {
                        log.info(" count = {}", count);
                        log.info("任务已全部完成！");
                        threadPool.shutdown();
                    }
                }
            });
        }
    }

    void increment() {
        lock.lock();
        try {
            count = count + 1;
        } finally {
            lock.unlock();
        }
    }


    /**
     * 声明一个阻塞队列（长度小于最大线程数maximumPoolSize 一般为 maximumPoolSize-1） 根据put方法（该方法是阻塞方法） 每执行完一个任务就从阻塞队列里移除一个标识(take方法)
     * 当阻塞队列满时 再次调用put方法此时会阻塞当前线程 直到有阻塞队列有数据移除为止
     * 这样保证了在不触发项目拒绝策略的情况下让现场一直运行
     */
    public void blockingQueueTask() {
        BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(9);
        ExecutorService threadPool = new ThreadPoolExecutor(5, 10,
                0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(10));
        int i = 0;
        while (i < 1000) {
            i++;
            int finalI = i;
            try {
                blockingQueue.put(i);
                log.info("blockingQueue.size() = {}", blockingQueue.size());
                threadPool.submit(new Runnable() {
                    @Override
                    public void run() {
                        log.info("当前任务 {}", finalI);
                        try {
                            blockingQueue.take();
                            Thread.sleep(2000);
                        } catch (InterruptedException e) {
                            log.error("err ", e);
                        }
                    }
                });
                System.out.println("blockingQueue.size():" + blockingQueue.size() + "  " + "threadpool:" + threadPool);

            } catch (InterruptedException e) {
                log.error("err ", e);
            }
        }
        threadPool.shutdown();
        log.info("任务已全部运行完成！");
    }
}